package org.apache.spark.streaming.rabbitmq;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.streaming.OutputMode;
import org.apache.spark.sql.streaming.StreamingQuery;
import org.apache.spark.sql.streaming.Trigger;
import scala.concurrent.duration.Duration;

import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

/**
 * <pre>
 *
 * Created by zhenqin.
 * User: zhenqin
 * Date: 2019/3/13
 * Time: 18:15
 * Vendor: yiidata.com
 *
 * </pre>
 *
 * @author zhenqin
 */
public class StructedStreamingJob implements Serializable {

    public static void main(String[] args) throws Exception {
        SparkSession spark = SparkSession.builder()
                .master("local[2]")
                .appName("AmqpFlowTest")
                .config("spark.sql.streaming.checkpointLocation", "file:///Volumes/Study/Work/IntelliJWork/spark-rabbitmq/data/checkpoint")
                //.enableHiveSupport()
                .getOrCreate();
        Map<String, String> amqpParams = new HashMap<>();
        amqpParams.put("amqp.host", "localhost");
        amqpParams.put("amqp.port", "5672");
        amqpParams.put("amqp.queueName", "testtopic");
        //amqpParams.put("amqp.exchangeName", "rabbitmq-exchange");
        //amqpParams.put("amqp.vHost", "/");
        amqpParams.put("amqp.username", "admin");
        amqpParams.put("amqp.password", "123456");

        SQLContext sqlContext = new SQLContext(spark.sparkContext());
        final Dataset<Row> kafka = sqlContext.readStream()
                .format("amqp")
                .schema(AmqpSource.amqpSchema())
                .options(amqpParams)
                .load();

        spark.listenerManager();

        StreamingQuery query = kafka.writeStream()
                .format("console")
                .outputMode(OutputMode.Append())
                .trigger(Trigger.ProcessingTime(Duration.apply(10, TimeUnit.SECONDS)))
                .start();
        System.out.println("开始启动 JavaStreaming 接受数据......");
        query.awaitTermination();
    }

}
